package com.share.device.emqx;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

//MQTT emqx服务器测试类
public class MqttDemo {

    public static void main(String[] args) {

        String subTopic = "testtopic/#"; //消息发布标识
        String pubTopic = "testtopic/1"; //消息订阅标识
        String content = "Hello World";  //消息内容

        int qos = 2; //消息级别：QoS 2：只有一次发送，确保消息只送达一次。

        String url = "tcp://192.168.100.130:1883"; //emqx服务器地址
        String clientId = "emqx_test";  //emqx服务器client标识

        MemoryPersistence memoryPersistence = new MemoryPersistence();


        try {

            //创建mqtt客户端
            MqttClient mqttClient = new MqttClient(url,clientId,memoryPersistence);

            //emqx连接对象
            MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
            mqttConnectOptions.setUserName("emqx_test");
            mqttConnectOptions.setPassword("emqx_test_password".toCharArray());

            // 保留会话
            mqttConnectOptions.setCleanSession(true);

            //mqtt连接到emqx
            System.out.println("Connecting to broker: " + url);
            mqttClient.connect(mqttConnectOptions);

            System.out.println("Connected");
            System.out.println("Publishing message: " + content);

            //设置消息发送后的回调函数对象
            mqttClient.setCallback(new MqttCallback() {

                @Override
                public void connectionLost(Throwable cause) {
                    // 连接丢失后，一般在这里面进行重连
                    System.out.println("连接断开，可以做重连");
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    // subscribe后得到的消息会执行到这里面
                    System.out.println("接收消息主题:" + topic);
                    System.out.println("接收消息Qos:" + message.getQos());
                    System.out.println("接收消息内容:" + new String(message.getPayload()));
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("deliveryComplete---------" + token.isComplete());
                }
            });

            //绑定发送的topic
            mqttClient.subscribe(subTopic);

            //发送消息
            MqttMessage mqttMessage = new MqttMessage(content.getBytes());
            mqttMessage.setQos(qos);
            mqttClient.publish(pubTopic,mqttMessage);
            System.out.println("Message published");


            mqttClient.disconnect();
            mqttClient.close();
            System.exit(0);





        } catch (MqttException e) {
            throw new RuntimeException(e);
        }

    }
}
